package zeebe.simple;

import io.zeebe.client.ZeebeClient;
import io.zeebe.client.ZeebeClientBuilder;
import io.zeebe.client.api.ZeebeFuture;
import io.zeebe.client.api.response.ActivatedJob;
import io.zeebe.client.api.response.DeploymentEvent;
import io.zeebe.client.api.response.Topology;
import io.zeebe.client.api.response.WorkflowInstanceEvent;
import io.zeebe.client.api.worker.JobClient;
import io.zeebe.client.api.worker.JobHandler;
import io.zeebe.client.api.worker.JobWorker;

import java.time.Duration;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Scanner;

public class App {

    public static void main(String[] args) {
        simpleDemo();
//        orderDemo();
//        processDemo();
//        nonBlockingDemo();
//        variablesPOJODemo();
//        topologyDemo();
    }

    public static void topologyDemo() {
        String broker = "localhost:26500";
        ZeebeClientBuilder builder = ZeebeClient.newClientBuilder().brokerContactPoint(broker).usePlaintext();

        try (ZeebeClient client = builder.build()) {
            System.out.println("Requesting topology with initial contact point " + broker);
            Topology topology = client.newTopologyRequest().send().join();
            System.out.println("Topology:");

            topology.getBrokers().forEach(b -> {
                System.out.println("    " + b.getAddress());
                b.getPartitions().forEach(p -> System.out.println("      " + p.getPartitionId() + " - " + p.getRole()));
            });
            System.out.println("Done.");
        }
    }

    public static void variablesPOJODemo() {
        String broker = "localhost:26500";
        ZeebeClientBuilder builder = ZeebeClient.newClientBuilder().brokerContactPoint(broker).usePlaintext();

        try (ZeebeClient client = builder.build()) {
            final Order order = new Order();
            order.setOrderId(31243);

            client.newCreateInstanceCommand()
                    .bpmnProcessId("demoProcess")
                    .latestVersion()
                    .variables(order)
                    .send()
                    .join();

            client.newWorker().jobType("foo").handler(new DemoJobHandler()).open();
            waitUntilSystemInput("exit");
        }
    }

    private static class DemoJobHandler implements JobHandler {
        @Override
        public void handle(final JobClient client, final ActivatedJob job) {
            final Order order = job.getVariablesAsType(Order.class);
            System.out.println("new job with orderId: " + order.getOrderId());
            order.setTotalPrice(46.50);
            client.newCompleteCommand(job.getKey()).variables(order).send();
        }
    }

    public static void nonBlockingDemo() {
        final String broker = "localhost:26500";
        final int numberOfInstances = 1_000;    //实际值就是1000。java7引入，分割数字增强可读性。
        final String bpmnProcessId = "demoProcess";
        final ZeebeClientBuilder builder = ZeebeClient.newClientBuilder().brokerContactPoint(broker).usePlaintext();

        try (ZeebeClient client = builder.build()) {
            System.out.println("Creating " + numberOfInstances + " workflow instances");

            final long startTime = System.currentTimeMillis();
            long instancesCreating = 0;

            //异步模式疯狂的创建流程实例
            while (instancesCreating < numberOfInstances) {
                ZeebeFuture<WorkflowInstanceEvent> future = client.newCreateInstanceCommand()
                        .bpmnProcessId(bpmnProcessId)
                        .latestVersion()
                        .send();
                //需要将future进行处理

                System.out.println("第" + instancesCreating + "个流程实例创建成功");
                instancesCreating++;
            }

            System.out.println("Took: " + (System.currentTimeMillis() - startTime));
        }
    }

    public static void processDemo() {
        final String broker = "localhost:26500";
        String bpmnProcessId = "demoProcess";
        String jobType = "foo";


        ZeebeClientBuilder clientBuilder = ZeebeClient.newClientBuilder().brokerContactPoint(broker).usePlaintext();
        try (ZeebeClient client = clientBuilder.build()) {

            DeploymentEvent deploymentEvent = client.newDeployCommand()
                    .addResourceFromClasspath("bpmn/demoProcess.bpmn")
                    .send()
                    .join();
            System.out.println("Deployment created with key: " + deploymentEvent.getKey());

            //部署流程和启动流程是独立的，可以分开
            System.out.println("Creating workflow instance");
            WorkflowInstanceEvent workflowInstanceEvent = client
                    .newCreateInstanceCommand()
                    .bpmnProcessId(bpmnProcessId)
                    .latestVersion()
                    .send()
                    .join();
            System.out.println("Workflow instance created with key: " + workflowInstanceEvent.getWorkflowInstanceKey());


            System.out.println("Opening job worker.");
            final JobWorker workerRegistration = client
                    .newWorker()
                    .jobType(jobType)
                    .handler(new ExampleJobHandler())
                    .timeout(Duration.ofSeconds(10))
                    .open();

            System.out.println("Job worker opened and receiving jobs.");


            waitUntilSystemInput("exit");
        }
    }

    /**
     * 实现jobworker，这个worker完成所有type=foo的工作
     */
    private static class ExampleJobHandler implements JobHandler {
        @Override
        public void handle(final JobClient client, final ActivatedJob job) {
            System.out.println(job);
            client.newCompleteCommand(job.getKey()).send().join();
        }
    }

    /**
     * 接收命令行退出命令，结束程序。输入exit
     *
     * @param exitCode
     */
    private static void waitUntilSystemInput(String exitCode) {
        try (Scanner scanner = new Scanner(System.in)) {
            while (scanner.hasNextLine()) {
                final String nextLine = scanner.nextLine();
                if (nextLine.contains(exitCode)) {
                    return;
                }
            }
        }
    }

    public static void orderDemo() {
        ZeebeClient client = connect();

        String bpmnPath = "bpmn/order.bpmn";
        deploy(client, bpmnPath);

        String processId = "order-process";
        Map<String, Object> data = new HashMap<>();
        data.put("orderId", 31243);
        data.put("orderItems", Arrays.asList(435, 182, 376));
        createInstance(client, processId, data);

        JobWorker jobWorker = client.newWorker()
                .jobType("payment-service")
                .handler((jobClient, job) -> {
                    //根据fectchVariables()获取指定的map,类似于过滤操作
                    Map<String, Object> variables = job.getVariablesAsMap();
                    System.out.println("Process order: " + variables.get("orderId"));

                    double price = 46.50;
                    System.out.println("Collect money: $" + price);
                    Map<String, Object> result = new HashMap<>();
                    result.put("totalPrice", price);

                    jobClient.newCompleteCommand(job.getKey())
                            .variables(result)
                            .send()
                            .join();
                })
                .fetchVariables("orderId")
                .open();

        JobWorker jobWorker2 = client.newWorker()
                .jobType("inventory-service")
                .handler((jobClient, job) -> {

                    System.out.println("第2个service执行完成");
                    jobClient.newCompleteCommand(job.getKey())
                            .send()
                            .join();
                })
                .open();

        JobWorker jobWorker3 = client.newWorker()
                .jobType("shipment-service")
                .handler((jobClient, job) -> {
                    System.out.println("第3个service执行完成");
                    jobClient.newCompleteCommand(job.getKey())
                            .send()
                            .join();
                })
                .open();

    }

    public static void simpleDemo() {
        ZeebeClient client = connect();
        System.out.println("Connected.");

        String bpmnPath = "bpmn/simple.bpmn";
        deploy(client, bpmnPath);

        String processId = "order-process";
        createInstance(client, processId, null);

        client.close();
        System.out.println("Closed.");
    }

    /**
     * connect ZeebeClient
     *
     * @return
     */
    public static ZeebeClient connect() {
        ZeebeClient client = ZeebeClient.newClientBuilder()
                .brokerContactPoint("localhost:26500")
                .usePlaintext()
                .build();
        return client;
    }

    /**
     * deploy workflow
     *
     * @param client
     * @param bpmnPath
     */
    public static void deploy(ZeebeClient client, String bpmnPath) {
        DeploymentEvent deployment = client.newDeployCommand()
                .addResourceFromClasspath(bpmnPath)
                .send()
                .join();
        int version = deployment.getWorkflows().get(0).getVersion();
        System.out.println("Workflow deployed. Version: " + version);
    }

    /**
     * create workflow instance
     *
     * @param client
     * @param processId
     */
    public static void createInstance(ZeebeClient client, String processId, Map map) {
        Map<String, Object> varMap = new HashMap<>();
        if (map != null) {
            varMap.putAll(map);
        }
        WorkflowInstanceEvent wfInstance = client.newCreateInstanceCommand()
                .bpmnProcessId(processId)
                .latestVersion()
                .variables(varMap)
                .send()
                .join();

        long workflowInstanceKey = wfInstance.getWorkflowInstanceKey();
        System.out.println("Workflow instance created. Key: " + workflowInstanceKey);
    }
}

class Order {
    private long orderId;
    private double totalPrice;

    public long getOrderId() {
        return orderId;
    }

    public void setOrderId(final long orderId) {
        this.orderId = orderId;
    }

    public double getTotalPrice() {
        return totalPrice;
    }

    public void setTotalPrice(final double totalPrice) {
        this.totalPrice = totalPrice;
    }
}

